package com.study.eg2;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

public class OrderedProducer {

  public static void main(String[] args)
      throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
    DefaultMQProducer producer = new DefaultMQProducer("GROUP_TEST");
    producer.setNamesrvAddr("localhost:9876");
    producer.start();
    producer.setRetryTimesWhenSendAsyncFailed(0);

    MessageQueueSelector selector = new MessageQueueSelector() {
      public MessageQueue select(List<MessageQueue> queues, Message message, Object o) {
        Integer id = (Integer) o;
        int index = id % queues.size();
        return queues.get(index);
      }
    };

    String[] tags = new String[]{"TagA", "TagB", "TagC"};
    List<Map<String, Object>> bizDatas = getBizDatas();
    for (int i = 0; i < bizDatas.size(); i++) {
      Map<String, Object> bizData = bizDatas.get(i);
      Message message = new Message("TopicTest", tags[i % tags.length], "" + bizData.get("msgType"),
          bizData.toString().getBytes());
      SendResult result = producer.send(message, selector, bizData.get("msgType"));
      System.out.printf("%s, body: %s%n", result, bizData);
    }

    producer.shutdown();
  }

  private static List<Map<String, Object>> getBizDatas() {
    List<Map<String, Object>> orders = new ArrayList<>();
    Map<String, Object> orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "存钱1000");
    orders.add(orderData);

    orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "取钱1000");
    orders.add(orderData);

    orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "存钱1000");
    orders.add(orderData);

    orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "存钱1000");
    orders.add(orderData);

    orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "存钱1000");
    orders.add(orderData);

    orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "取钱1000");
    orders.add(orderData);

    orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "取钱1000");
    orders.add(orderData);

    orderData = new HashMap<>();
    orderData.put("msgType", 1001);
    orderData.put("userId", "tony");
    orderData.put("desc", "存钱1000");
    orders.add(orderData);
    return orders;
  }

}
